package com.sharp.mq.config;

import lombok.Builder;
import lombok.Data;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.FactoryBean;

/**
 * Title: DynamicConsumerContainerFactory
 * Description: Copyright: Copyright (c) 2019 Company: BHFAE
 *
 * @author Sharp
 * @date 2021/7/5 10:01
 */
@Data
@Builder
public class DynamicConsumerContainerFactory implements FactoryBean<SimpleMessageListenerContainer> {

    private boolean autoAck;

    private Exchange exchange;
    private Queue queue;
    private Binding binding;

    private ConnectionFactory connectionFactory;
    private RabbitAdmin rabbitAdmin;
    private Integer concurrentNum;

    private MessageDelegate messageDelegate;

    @Override
    public SimpleMessageListenerContainer getObject() {
        check();

        rabbitAdmin.declareQueue(queue);
        rabbitAdmin.declareExchange(exchange);
        rabbitAdmin.declareBinding(binding);

        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueues(queue);
        container.setPrefetchCount(20);
        container.setConcurrentConsumers(concurrentNum == null ? 1 : concurrentNum);
        container.setAcknowledgeMode(autoAck ? AcknowledgeMode.AUTO : AcknowledgeMode.MANUAL);
        container.setExposeListenerChannel(true);
        container.setDefaultRequeueRejected(false);
        container.setErrorHandler(new MyConditionalRejectingErrorHandler());
        Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
        MyMessageListenerAdapter adapter = new MyMessageListenerAdapter(messageDelegate, converter);
        container.setMessageListener(adapter);
        return container;
    }

    private void check() {
        if (null == rabbitAdmin || null == connectionFactory) {
            throw new IllegalArgumentException("rabbitAdmin and connectionFactory should not be null!");
        }
    }

    @Override
    public Class<?> getObjectType() {
        return SimpleMessageListenerContainer.class;
    }
}
